package tarns_result.jiaoyu

import common.Tools.getSC
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

import scala.collection.mutable
import scala.util.parsing.json.JSON

object JiaoYu_CSV {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = getSC("JiaoYu")
//    val input_original = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/EducationStatistics_ToJson/part-00000"
        val input_original = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/WorldDevelopmentIndicators_ToJson/part-00000"

    val original_lines = sc.textFile(input_original)
    val riginal_rdd: RDD[Original] = toOriginalRDD(original_lines)
    val riginal_map = riginal_rdd.map(r => (r.INDICATOR_CNKI, r))

    val input_match = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/match"
    val match_lines = sc.textFile(input_match)
    val match_rdd: RDD[Match] = toMatchRDD(match_lines)
    val match_map = match_rdd.map(m => (m.InternationalIndicator, m))

    val join_rdd: RDD[(String, (Original, Option[Match]))] = riginal_map.leftOuterJoin(match_map)
    val res: RDD[String] = join_rdd.map {
      case (indicator, (r, m)) => {
        val the_r: Original = r
        val the_m: Match = m.getOrElse(null)
        val r_str = the_r.toString
        val m_str = if (the_m == null) "@@" else the_m.toString
        r_str + m_str
      }
    }

//    val outpath="/Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/EducationStatistics_JIAOYU"
        val outpath="/Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/WorldDevelopmentIndicators_JIAOYU"
    //    res.foreach(println(_))
    res.repartition(1).saveAsTextFile(outpath)
    sc.stop()
  }


  def toOriginalRDD(json_lines: RDD[String]): RDD[Original] = {
    val maps: RDD[Option[Any]] = json_lines.map(josn_line => JSON.parseFull(josn_line))
    maps.map {
      val my_map = new mutable.HashMap[String, String]()
      r =>
        r match {
          case Some(map: Map[String, Any]) => {
            map.foreach(x => my_map += (x._1 -> x._2.toString))
            val YEAR = if (my_map.getOrElse("Year", "") != "") {
              my_map.getOrElse("Year", "") + "年"
            } else {
              ""
            }
            val REPORTGROUPER = YEAR + "€" + my_map.getOrElse("ReporterCode", "") + "€" + my_map.getOrElse("Indicator", "") //这3个字段肯定不为空,IBRDchange这个代码过滤了空的
              Original(YEAR, my_map.getOrElse("Indicator", ""), my_map.getOrElse("Reporter", ""),
              my_map.getOrElse("ReporterCode", ""), my_map.getOrElse("ObservationValue", ""), my_map.getOrElse("Unit", ""),
              my_map.getOrElse("SourceName", ""), "N2020WBDATA", REPORTGROUPER, "")
          }
        }
    }
  }

  def toMatchRDD(lines: RDD[String]): RDD[Match] = {
    val arrays: RDD[Array[String]] = lines.map(line => line.split("\t"))
    val filter_arrays: RDD[Array[String]] = arrays.filter(arr => if (arr.length == 5) true else false)
    //=4的缺少InternationalIndicator，无法关联，直接过滤掉
    filter_arrays.map(arr => Match(arr(4), arr(2), arr(3)))
  }

}

case class Original(YEAR:String,INDICATOR_CNKI:String,AREA:String,AREACODE:String,NUMBER:String,UNIT:String,TITLENAME:String,YEARBOOKNAME:String,REPORTGROUPER:String,OTHERINFO:String){
  override def toString: String ={
    YEAR+"@"+INDICATOR_CNKI+"@"+AREA+"@"+AREACODE+"@"+NUMBER+"@"+UNIT+"@"+TITLENAME+"@"+YEARBOOKNAME+"@"+REPORTGROUPER+"@"+OTHERINFO
  }
}
case class Match(InternationalIndicator:String,INDICATORCODE:String,INDICATORNAME:String){
  override def toString: String = {
    "@"+INDICATORCODE+"@"+INDICATORNAME
  }
}